package com.jintian.smart.kernel.switching.runtime;

import com.jintian.smart.kernel.switching.annotation.ExtractJob;
import com.jintian.smart.kernel.switching.annotation.ExtractJobVariables;
import com.jintian.smart.kernel.switching.entity.ETLExtractJob;
import com.jintian.smart.kernel.switching.mapper.IETLExtractJobMapper;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.Signature;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.core.annotation.Order;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.event.TransactionPhase;
import org.springframework.transaction.event.TransactionalEventListener;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.Assert;

import java.lang.reflect.Method;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.Map;
import java.util.Stack;

@Aspect
public class ExtractJobAspect implements ApplicationContextAware {
    private static final Map<String, Object> EMPTY = Collections.emptyMap();
    private ApplicationEventPublisher aplicationEventPublisher;

    @Autowired
    private IETLExtractJobMapper etlExtractJobMapper;

    @Autowired
    @Qualifier("transactionManager")
    private PlatformTransactionManager transactionManager;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        ApplicationContext p = applicationContext;
        // 顶级ApplicationContext触发消息
        while (p.getParent() != null) {
            p = p.getParent();
        }
        this.aplicationEventPublisher = p;

    }

    private static final ThreadLocal<Stack<Map<String, Object>>> etlExtractJobMap = new ThreadLocal<Stack<Map<String, Object>>>() {
        @Override
        protected Stack<Map<String, Object>> initialValue() {
            return new Stack<>();
        }
    };

    /**
     * 绑定采集任务的执行变量
     *
     * @param item
     */
    public static void bindExtractJobVariables(Map<String, Object> item) {
        if (item == null) {
            etlExtractJobMap.get().push(EMPTY);
        } else {
            etlExtractJobMap.get().push(item);
        }
    }

    @Order(20)
    @Around("@annotation(com.jintian.smart.kernel.switching.annotation.ExtractJob)")
    public Object aroundExtractJob(ProceedingJoinPoint point) throws Throwable {
        Signature signature = point.getSignature();
        MethodSignature methodSignature = (MethodSignature) signature;
        Method method = methodSignature.getMethod();
        ExtractJob extractJob = method.getAnnotation(ExtractJob.class);
        int cnt = etlExtractJobMap.get().size();
        Object ret = point.proceed();
        Assert.isTrue(cnt + 1 == etlExtractJobMap.get().size(), "未绑定采集任务的执行变量");
        Map<String, Object> variables = etlExtractJobMap.get().pop();

        if (TransactionSynchronizationManager.isSynchronizationActive()
                && TransactionSynchronizationManager.isActualTransactionActive()) {
            if (EMPTY != variables) {
                aplicationEventPublisher.publishEvent(new AfterExtractJobTransactionEvent(extractJob, variables));
            }
        } else {
            this.onExtractJob(new AfterExtractJobTransactionEvent(extractJob, variables));
        }
        return ret;
    }

    @Order(10)
    @Around("@annotation(com.jintian.smart.kernel.switching.annotation.ExtractJobVariables)")
    @SuppressWarnings("unchecked")
    public Object ExtractJobVariables(ProceedingJoinPoint point) throws Throwable {
        Signature signature = point.getSignature();
        MethodSignature methodSignature = (MethodSignature) signature;
        Method method = methodSignature.getMethod();
        ExtractJobVariables extractJobVariables = method.getAnnotation(ExtractJobVariables.class);
        Map<String, Object> item = (Map<String, Object>) point.proceed();
        if (extractJobVariables.job().length > 0) {
            if (item != null) {
                for (ExtractJob job : extractJobVariables.job()) {
                    if (TransactionSynchronizationManager.isSynchronizationActive() && TransactionSynchronizationManager.isActualTransactionActive()) {
                        aplicationEventPublisher.publishEvent(new AfterExtractJobTransactionEvent(job, item));
                    } else {
                        this.onExtractJob(new AfterExtractJobTransactionEvent(job, item));
                    }
                }
            }
        } else {
            if (item == null) {
                etlExtractJobMap.get().push(EMPTY);
            } else {
                etlExtractJobMap.get().push(item);
            }
        }
        return item;
    }

    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    private void onExtractJob(AfterExtractJobTransactionEvent event) {
        TransactionTemplate template = new TransactionTemplate(transactionManager);
        template.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);

        ExtractJob extract = (ExtractJob) event.getSource();
        ETLExtractJob job = new ETLExtractJob();
        job.setCreateTime(LocalDateTime.now());
        job.setDataType(extract.dataType());
        job.setRetries(0);
        job.setRuleName(extract.rule());
        job.setStatus(ETLExtractJob.Status.waiting.name());
        job.setVariables(event.getVariables());
        template.executeWithoutResult((status) -> {
            etlExtractJobMapper.insert(job);
        });

    }

    static class AfterExtractJobTransactionEvent extends ApplicationEvent {
        private static final long serialVersionUID = 1L;
        private final Map<String, Object> variables;

        public AfterExtractJobTransactionEvent(ExtractJob source, Map<String, Object> variables) {
            super(source);
            this.variables = variables;
        }

        public Map<String, Object> getVariables() {
            return variables;
        }
    }
}
